package buka.oversea.node4.model;

import buka.oversea.node4.config.RaftConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

@Component
public class RaftNode {
    // 节点状态
    public enum State {FOLLOWER, CANDIDATE, LEADER}

    private final String nodeId;
    private State state = State.FOLLOWER;
    private int currentTerm = 0;
    private String votedFor = null;
    private String leaderId = null;

    // Raft日志
    private List<LogEntry> log = Collections.synchronizedList(new ArrayList<>());
    private int commitIndex = -1;
    private int lastApplied = -1;

    // 其他节点信息
    private final List<String> peerNodes;
    private final Map<String, Integer> nextIndex = new ConcurrentHashMap<>();
    private final Map<String, Integer> matchIndex = new ConcurrentHashMap<>();

    // 状态机 (账户余额)
    private final Map<String, Integer> accountBalances = new ConcurrentHashMap<>();

    // 定时器
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private ScheduledFuture<?> electionTimeout;
    private ScheduledFuture<?> heartbeatTimeout;

    // 投票请求线程池
    private final ExecutorService voteExecutor = Executors.newScheduledThreadPool(5);
    // 投票请求线程池
    private final ExecutorService heartBeatExecutor = Executors.newScheduledThreadPool(5);
    // 投票显示锁
    private final ReentrantLock raftLock = new ReentrantLock();  // 显示锁，可重入，公平策略可选

    // 通信组件
    private final RestTemplate restTemplate;

    @Autowired
    public RaftNode(RaftConfig raftConfig, RestTemplate restTemplate) {
        this.nodeId = raftConfig.getNodeId();
        this.peerNodes = raftConfig.getPeers();
        this.restTemplate = restTemplate;
        resetElectionTimer();
    }

    // RPC处理：请求投票
    public synchronized VoteResult requestVote(int term, String candidateId, int lastLogIndex, int lastLogTerm) {
        if (term < currentTerm) return new VoteResult(currentTerm, false);

        if (term > currentTerm) {
            currentTerm = term;
            stepDown();
        }

        boolean voteGranted = (votedFor == null || votedFor.equals(candidateId)) && isLogUpToDate(lastLogIndex, lastLogTerm);

        if (voteGranted) {
            votedFor = candidateId;
            resetElectionTimer(); // 投票完成之后，为了防止自己马上又超时，顺手调用了 resetElectionTimer()——这只是把“下一次选举超时”往后推迟，并不是立即把自己变成 Candidate 去抢选票。
        }
        return new VoteResult(currentTerm, voteGranted);
    }

    // 检查日志是否最新
    private boolean isLogUpToDate(int lastLogIndex, int lastLogTerm) {
        int myLastLogIndex = log.size() - 1;
        int myLastLogTerm = (myLastLogIndex >= 0) ? log.get(myLastLogIndex).term : 0;

        return (lastLogTerm > myLastLogTerm) || (lastLogTerm == myLastLogTerm && lastLogIndex >= myLastLogIndex);
    }

    // RPC处理：追加条目
    public synchronized AppendResult appendEntries(int term, String leaderId, int prevLogIndex, int prevLogTerm, List<LogEntry> entries, int leaderCommit) {
        if (term < currentTerm) return new AppendResult(currentTerm, false);

        resetElectionTimer();
        if (term > currentTerm) {
            currentTerm = term;
            stepDown();
        }
        this.leaderId = leaderId;

        // 日志一致性检查
        if (prevLogIndex >= 0) {
            if (log.size() <= prevLogIndex) {
                return new AppendResult(currentTerm, false);
            }
            if (log.get(prevLogIndex).term != prevLogTerm) {
                return new AppendResult(currentTerm, false);
            }
        }

        // 追加新日志
        if (entries != null && !entries.isEmpty()) {
            // 删除冲突的日志（如果有）
            int newIndex = prevLogIndex + 1;
            if (log.size() > newIndex) {
                log = new ArrayList<>(log.subList(0, newIndex));
            }
            log.addAll(entries);
        }

        // 更新提交索引
        if (leaderCommit > commitIndex) {
            commitIndex = Math.min(leaderCommit, log.size() - 1);
            applyCommittedEntries();
        }

        return new AppendResult(currentTerm, true);
    }

    // 应用已提交的日志到状态机
    private void applyCommittedEntries() {
        while (lastApplied < commitIndex) {
            lastApplied++;
            LogEntry entry = log.get(lastApplied);
            applyOperation(entry.operation);
        }
    }

    // 执行余额操作
    private void applyOperation(Operation op) {
        accountBalances.compute(op.account, (k, v) -> {
            int current = (v == null) ? 0 : v;
            return switch (op.type) {
                case "add" -> current + op.amount;
                case "sub" -> current - op.amount;
                default -> current;
            };
        });
    }

    /**
     * 「选举」并不是只在初始化时触发一次，而是靠选举超时定时器反复启动的：
     * 初始化时确实会调一次 resetElectionTimer()，这是第一次倒计时。
     * 之后只要还是 Follower / Candidate，每次收到合法 Leader 的心跳、或给别人投票、或自己刚投完票，都会再调一次 resetElectionTimer()，把倒计时重新拨回 150~300 ms。
     * 一旦Leader 失联（心跳停了），resetElectionTimer() 里的定时任务不会再被重置，于是 150~300 ms 后超时触发 startElection()，开始新一轮选举。
     * 代码里看不到“Leader 下线 → 立即选举”的显式调用，是因为选举触发逻辑隐藏在超时定时器里
     */
    private void resetElectionTimer() {
        if (electionTimeout != null) electionTimeout.cancel(false); // 如果之前有一个选举超时任务，就取消它
        electionTimeout = scheduler.schedule(() -> {
            synchronized (this) {
                if (state != State.LEADER) {
                    startElection();
                }
            }
        }, 150 + new Random().nextInt(150), TimeUnit.MILLISECONDS);  // 创建一个新的定时任务，延迟150-300ms之间的随机时间后执行，使用随机时间是为了避免多个节点同时发起选举，减少“投票分裂”问题
    }

    // 开始选举
    private void startElection() {
        raftLock.lock();
        try {
            // 以下三步为原子方法，raft论文规定必须要一次性完成
            state = State.CANDIDATE;
            currentTerm++;
            votedFor = nodeId;
            // 票数计数器,先给自己一票
            AtomicInteger votes = new AtomicInteger(1);
            // 请求其他节点投票
            for (String peer : peerNodes) {
                voteExecutor.submit(() -> {
                    int lastLogIndex = log.size() - 1;
                    int lastLogTerm = (lastLogIndex >= 0) ? log.get(lastLogIndex).term : 0;
                    VoteResult result = sendVoteRequest(peer, currentTerm, nodeId, lastLogIndex, lastLogTerm);
                    synchronized (RaftNode.this) {
                        if (state != State.CANDIDATE) return;
                        //发现更高的任期，立即准备下台，降级为follower
                        if (result.term > currentTerm) {
                            currentTerm = result.term;
                            stepDown();
                            return;
                        }
                        if (result.voteGranted) {
                            if (votes.incrementAndGet() > (peerNodes.size() + 1) / 2) {
                                becomeLeader();
                            }
                        }
                    }
                });

            }
        } catch (Exception e) {
            // 网络异常 / 序列化异常 / 超时等
        } finally {
            // 释放锁
            raftLock.unlock();
            resetElectionTimer();
        }
    }

    // 成为Leader
    private void becomeLeader() {
        state = State.LEADER;
        leaderId = nodeId;
        for (String peer : peerNodes) {
            // Leader 认为每个 follower 下一条该从 log.size() 开始复制。
            nextIndex.put(peer, log.size());
            // 当前确认已复制到该 follower 的最高日志索引，初始化为 –1（表示尚未复制任何条目）。
            matchIndex.put(peer, -1);
        }
        System.out.println("当前leader为：" + leaderId);
        sendHeartbeats();
        if (heartbeatTimeout != null) heartbeatTimeout.cancel(false);
        heartbeatTimeout = scheduler.scheduleAtFixedRate(this::sendHeartbeats, 0, 50, TimeUnit.MILLISECONDS); // leader 存活，定时器一直跑
    }

    // 并非单纯的心跳，而是leader周期性的向所有 follower 发送「AppendEntries RPC」，它同时承担了 心跳保活 和 日志复制 两项职责
    private void sendHeartbeats() {
        for (String peer : peerNodes) {
            heartBeatExecutor.submit(() -> {
                int nextIdx = nextIndex.getOrDefault(peer, 0);
                List<LogEntry> entries = null;

                if (log.size() > nextIdx) {
                    entries = new ArrayList<>(log.subList(nextIdx, log.size()));
                }

                int prevLogIndex = nextIdx - 1;
                int prevLogTerm = (prevLogIndex >= 0 && prevLogIndex < log.size()) ? log.get(prevLogIndex).term : 0;

                AppendResult result = sendAppendEntries(peer, currentTerm, nodeId, prevLogIndex, prevLogTerm, entries, commitIndex);

                synchronized (RaftNode.this) {
                    if (result.term > currentTerm) {
                        currentTerm = result.term;
                        stepDown();
                        return;
                    }
                    if (result.success) {
                        nextIndex.put(peer, log.size());
                        matchIndex.put(peer, log.size() - 1);
                        updateCommitIndex();
                    } else if (nextIdx > 0) {
                        nextIndex.put(peer, nextIdx - 1);
                    }
                }
            });
        }
    }

    // 更新提交索引
    private void updateCommitIndex() {
        List<Integer> matchIndices = new ArrayList<>(matchIndex.values());
        matchIndices.add(log.size() - 1); // 包括自己
        Collections.sort(matchIndices);
        Collections.reverse(matchIndices);

        int n = matchIndices.get((peerNodes.size() + 1) / 2);
        if (n > commitIndex && log.get(n).term == currentTerm) {
            commitIndex = n;
            applyCommittedEntries();
        }
    }

    // 降级为Follower
    private void stepDown() {
        state = State.FOLLOWER;
        leaderId = null;
        votedFor = null;
        resetElectionTimer();
        if (heartbeatTimeout != null) {
            heartbeatTimeout.cancel(false);
            heartbeatTimeout = null;
        }
    }

    // 获取账户余额
    public int getBalance(String account) {
        return accountBalances.getOrDefault(account, 0);
    }

    public String getNodeId() {
        return nodeId;
    }

    // 添加新操作
    public synchronized boolean addOperation(Operation op) {
        if (state != State.LEADER) return false;

        log.add(new LogEntry(currentTerm, op));
        return true;
    }

    // RPC请求封装
    private VoteResult sendVoteRequest(String peer, int term, String candidateId, int lastLogIndex, int lastLogTerm) {
        try {
            String url = "http://" + peer + "/api/balance/raft/requestVote";

            // 创建请求对象
            Map<String, Object> request = new HashMap<>();
            request.put("term", term);
            request.put("candidateId", candidateId);
            request.put("lastLogIndex", lastLogIndex);
            request.put("lastLogTerm", lastLogTerm);

            // 发送POST请求
            ResponseEntity<Map> response = restTemplate.postForEntity(url, request, Map.class);

            if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
                Map<String, Object> body = response.getBody();
                return new VoteResult(((Number) body.get("term")).intValue(), (Boolean) body.get("voteGranted"));
            }
        } catch (Exception e) {
            System.err.println("[" + nodeId + "] RequestVote RPC to " + peer + " failed: " + e.getMessage());
        }
        return new VoteResult(term, false);
    }

    private AppendResult sendAppendEntries(String peer, int term, String leaderId, int prevLogIndex, int prevLogTerm, List<LogEntry> entries, int leaderCommit) {
        try {
            String url = "http://" + peer + "/api/balance/raft/appendEntries";

            // 创建请求对象
            Map<String, Object> request = new HashMap<>();
            request.put("term", term);
            request.put("leaderId", leaderId);
            request.put("prevLogIndex", prevLogIndex);
            request.put("prevLogTerm", prevLogTerm);
            request.put("leaderCommit", leaderCommit);

            // 转换日志条目
            if (entries != null) {
                List<Map<String, Object>> entriesList = new ArrayList<>();
                for (LogEntry entry : entries) {
                    Map<String, Object> entryMap = new HashMap<>();
                    entryMap.put("term", entry.term);

                    Map<String, Object> opMap = new HashMap<>();
                    opMap.put("type", entry.operation.type);
                    opMap.put("account", entry.operation.account);
                    opMap.put("amount", entry.operation.amount);

                    entryMap.put("operation", opMap);
                    entriesList.add(entryMap);
                }
                request.put("entries", entriesList);
            } else {
                request.put("entries", null);
            }

            // 发送POST请求
            ResponseEntity<Map> response = restTemplate.postForEntity(url, request, Map.class);

            if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
                Map<String, Object> body = response.getBody();
                return new AppendResult(((Number) body.get("term")).intValue(), (Boolean) body.get("success"));
            }
        } catch (Exception e) {
            System.err.println("[" + nodeId + "] AppendEntries RPC to " + peer + " failed: " + e.getMessage());
        }
        return new AppendResult(term, false);
    }

    // 获取状态
    public State getState() {
        return state;
    }

    // 获取Leader ID
    public String getLeaderId() {
        return leaderId;
    }

    // 数据结构定义
    public static class LogEntry {
        // （日志条目的任期号）是 Raft 用来**判断两条日志“谁更新”的核心字段，也是保证安全性（safety）**的关键。没有它，仅靠 log index 无法区分“旧数据”和“新数据”。
        //  必须带着term去投票/复制：只有拥有“至少和自己一样新”日志的 Candidate，才能赢得投票、只有让 Follower 先校验“之前那一条日志的 term 与 index”都匹配，才能继续追加新日志
        public int term;
        public Operation operation;

        public LogEntry() {
        }

        public LogEntry(int term, Operation operation) {
            this.term = term;
            this.operation = operation;
        }
    }

    public static class Operation {
        public String type; // "add" or "sub"
        public String account;
        public int amount;

        public Operation() {
        }

        public Operation(String type, String account, int amount) {
            this.type = type;
            this.account = account;
            this.amount = amount;
        }
    }

    public record VoteResult(int term, boolean voteGranted) {
    }

    public record AppendResult(int term, boolean success) {
    }
}
